package com.atguigu.hbase;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.ColumnValueFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

public class HbaseDML {

    public static void main(String[] args) throws IOException {
        putData("gmall","user","9901","info","name","linghuchong");
        putData("gmall","user","9901","info","name","yuebuqun");
        putData("gmall","user","9901","info","age","44");
        putData("gmall","user","9902","info","name","dongfang");
        putData("gmall","user","9902","info","age","55");
        putData("gmall","user","9903","info","name","renwoxing");
        putData("gmall","user","9903","info","age","66");
        putData("gmall","user","9904","info","name","abc1");
        putData("gmall","user","9904","info","age","77");
        putData("gmall","user","9904","info","name","abc2");
        putData("gmall","user","9904","info","age","88");

        putData("gmall","user","9905","info","name","test001");
       // getData("gmall","user","9901","info","name");
        scanData("gmall","user","9901","9909");

      //  deleteData("gmall","user","9904","info","name");

        System.out.println("ok");
    }

    static void putData(String namespace,String tablename,String rowkey,String cf,String col,String value) throws IOException {
        Connection connection = HbaseConnect.getConnection();
        TableName tableName = TableName.valueOf(namespace, tablename);
        Table table = connection.getTable(tableName);

        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(Bytes.toBytes(cf),Bytes.toBytes(col),Bytes.toBytes(value));
        table.put(put);

        table.close();


    }

    static void getData(String namespace,String tablename,String rowkey,String cf,String col) throws IOException {
        Connection connection = HbaseConnect.getConnection();
        TableName tableName = TableName.valueOf(namespace, tablename);
        Table table = connection.getTable(tableName);

        Get get = new Get(Bytes.toBytes(rowkey));
        get.addColumn(Bytes.toBytes(cf),Bytes.toBytes(col));
        // get.readAllVersions();

        Result result = table.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println("cell = " + Bytes.toString( CellUtil.cloneValue(cell))  );
        }

    }


    static  void scanData(String namespace,String tablename,String startrow,String stoprow) throws IOException {
        Connection connection = HbaseConnect.getConnection();
        TableName tableName = TableName.valueOf(namespace, tablename);
        Table table = connection.getTable(tableName);

        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes(startrow));
        scan.withStopRow(Bytes.toBytes(stoprow));

        //加入列值过滤,只能列出来某一列 进行过滤
       // scan.addColumn(Bytes.toBytes("info"),Bytes.toBytes("age"));
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
//        ColumnValueFilter columnValueFilter = new ColumnValueFilter(
//                // 列族
//                Bytes.toBytes("info"),
//                // 列名
//                Bytes.toBytes("age"),
//                // 匹配规则  一般为相等  也可以是大于等于 小于等于
//                CompareOperator.GREATER,
//                Bytes.toBytes("60")
//        );
//        filterList.addFilter(columnValueFilter);

        //单值过滤器 可以显示其他列的值
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                 Bytes.toBytes("info"),
                // 列名
                Bytes.toBytes("age"),
                // 匹配规则  一般为相等  也可以是大于等于 小于等于
                CompareOperator.GREATER,
                Bytes.toBytes("60")
        );
        //
        singleColumnValueFilter.setFilterIfMissing(true);
        filterList.addFilter(singleColumnValueFilter);

        scan.setFilter(filterList);


        ResultScanner scanner = table.getScanner(scan);

        for (Result result : scanner) {
            Cell[] cells = result.rawCells();
            for (Cell cell : cells) {
                System.out.println("cell = " + Bytes.toString(CellUtil.cloneValue(cell)) );
            }
        }

    }


    static void deleteData(String namespace,String tablename,String rowkey,String cf,String col) throws IOException {
        Connection connection = HbaseConnect.getConnection();
        TableName tableName = TableName.valueOf(namespace, tablename);
        Table table = connection.getTable(tableName);


        Delete delete = new Delete(Bytes.toBytes(rowkey));
        if(cf!=null&&col!=null){
           delete.addColumn(Bytes.toBytes(cf),Bytes.toBytes(col));
        }
        table.delete(delete);


    }
}
